import asyncio
import json
import logging
import pymongo
import pytz
import random
import string
import threading
import requests
import qianfan
import os
import time
import websockets
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import timedelta
#
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from smtplib import SMTP_SSL

import paho.mqtt.client as mqtt  # mqtt
import pika
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.date import DateTrigger
from flask import Flask, redirect, url_for, request, jsonify
from flask_jwt_extended import create_access_token, JWTManager, create_refresh_token, \
    jwt_required
from flask_login import UserMixin, logout_user, login_required
from motor.motor_asyncio import AsyncIOMotorClient
from werkzeug.security import generate_password_hash, check_password_hash

url = "" #url
# MQTT服务器地址和端口
MQTT_SERVER = ""#服务地址
MQTT_PORT = 1883 #端口
# MQTT主题
MQTT_TOPIC_down = ""#相应的主题
MQTT_TOPIC = ""#相关mqtt主题
MQTT_client_id = ""#连接id
insert_template = ("_id:{event_time} ,device_id:{device_id},shuishen:{shuishen} ,shidu:{shidu}, turangshidu:{"
                   "turangshidu},wendu:{wendu} , yudi:{yudi},light:{light},fengshan:{fengshan},shuibeng:{shuibeng},"
                   "fengmingqi:{fengmingqi},data_time:{data_time}")
tz = pytz.timezone('Asia/Shanghai')
publish_template = {
    "services": [{
        "service_id": "jieshou",
        "properties": {
            "fengshan": "OFF",
            "shuibeng": "OFF",
            "fengmingqi": "OFF"
        }
    }
    ]
}
OFF = "OFF"
ON = "ON"
find_time_query = "key_time:{key_time},start_time:{start_time},end_time:{end_time}"

os.environ["QIANFAN_ACCESS_KEY"] = ""#千帆模型的相应配置
os.environ["QIANFAN_SECRET_KEY"] = ""

chat_comp = qianfan.ChatCompletion()


def get_baidu_token():
    url = ""#千帆模型的相关url

    payload = json.dumps("")

    headers = {

        'Content-Type': 'application/json',

        'Accept': 'application/json'

    }

    response = requests.request("POST", url, headers=headers, data=payload)

    return response.json().get("access_token")


def get_baidu_response_json(content):
    url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions?access_token=" + get_baidu_token()
    payload = json.dumps({
        "messages": [
            {
                "role": "user",
                "content": content
            }
        ]
    })
    headers = {
        'Content-Type': 'application/json'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    return response.text


def get_baidu_json_make(reponse):
    content = json.loads(reponse)
    return content["result"]


# 指定特定模型
# resp = chat_comp.do(model="ERNIE-Bot", messages=[{
#     "role": "user",
#     "content": "你好"
# }])
#
# print(resp)


class logging_different():  # 写日志
    def __init__(self, name, log_level=logging.DEBUG, handler_level=logging.DEBUG, filename=None):
        self.name = name
        self.logger = logging.getLogger(name)  # 记录日志消息
        self.logger.setLevel(log_level)
        self.handler = logging.StreamHandler()  # 日志输出
        self.handler.setLevel(handler_level)
        self.formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
        self.handler.setFormatter(self.formatter)
        self.logger.addHandler(self.handler)
        self.filters = []  # 过滤信息
        if filename:
            self.add_file_handler(filename)

    def get_logger(self):
        return self.logger

    def set_level(self, level):
        self.logger.setLevel(level)
        self.handler.setLevel(level)

    def set_formatter(self, formatter):
        self.formatter = formatter
        self.handler.setFormatter(formatter)

    def add_handler(self, handler):
        self.logger.addHandler(handler)

    def add_filter(self, filter):
        self.filters.append(filter)
        self.handler.addFilter(filter)

    def remove_filter(self, filter):
        if filter in self.filters:
            self.filters.remove(filter)
            self.handler.removeFilter(filter)

    def add_file_handler(self, log_file):
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(self.formatter)
        self.logger.addHandler(file_handler)


class NoDataWebSocketFilter(logging.Filter):
    def filter(self, record):
        if "PING '' [0 bytes]" in record.getMessage() or "PONG '' [0 bytes]" in record.getMessage():
            return True
        return False


flask_logger = logging_different("flask", filename="flask.log")
# websocket_logger = logging_different("websockets", filename="websockets.log")
# websocket_logger.add_filter(NoDataWebSocketFilter())
mongodb_logger = logging_different("mongodb", filename="mongodb.log")


class Mqtt_client_test():
    def __init__(self, server=MQTT_SERVER, port=MQTT_PORT, topic=MQTT_TOPIC_down, client_id=MQTT_client_id,
                 username="", password=""):
        self.server = server
        self.port = port
        self.topic = topic
        self.client_id = client_id
        self.username = username
        self.password = password
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id)
        self.client.on_connect = self.on_connect
        self.client.on_publish = self.on_publish
        self.client.on_subscribe = self.on_subscribe
        self.client.on_message = self.on_msg

    def on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            print(f"M,Failed to connect: {reason_code}. loop_forever() will retry connection")
        else:
            print(f"M,Connected successfully: {reason_code}")

    def on_publish(self, client, userdata, mid, reason_code, properties):
        print(f"M,Published message: {userdata}", "mid:", mid, "reason_code:", reason_code, "properties:", properties)

    def on_subscribe(self, client, userdata, mid, reason_code_list, properties):
        if reason_code_list[0].is_failure:
            print(f"M,Broker rejected you subscription: {reason_code_list[0]}")
        else:
            print(f"M,Broker granted the following QoS: {reason_code_list[0].value}")

    def on_msg(self, client, userdata, msg):
        print("M,message:", msg.topic + " " + str(msg.payload))

    def connect(self):
        self.client.username_pw_set(self.username, self.password)
        self.client.connect(self.server, self.port)
        self.client.loop_start()

    def publish(self, message, topic, template=publish_template):
        self.client.publish(topic, message)

    def disconnect(self):
        self.client.loop_stop()
        self.client.disconnect()


class Amqp_accept():  # amqp类
    def __init__(self, amqp_host, amqp_port, amqp_queue_name, virtual_host, amqp_username, amqp_password):
        self.host = amqp_host
        self.port = amqp_port
        self.queue_name = amqp_queue_name
        self.virtual_host = virtual_host
        self.username = amqp_username
        self.password = amqp_password
        self.consumer_tag = None

        self.stop_event = threading.Event()  # 创建一个事件，用于通知线程停止
        self.receive_thread = None
        self.stop = None

    def connect(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(self.host, self.port, self.virtual_host,
                                                                            credentials=pika.PlainCredentials(
                                                                                self.username, self.password)))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=self.queue_name, durable=True)

        print(f"A,Connected to amqp on {self.host}:{self.port}, listening on queue {self.queue_name}")

    def start_accept(self):
        def callback(ch, method, properties, body):
            # self.channel.basic_ack(delivery_tag=)
            # print('[x] Received %r' % str(body))
            # print(body[2:-1])
            # print(str(body))
            dict_data = json.loads(str(body)[2:-1])
            # print(dict_data)
            event_time = dict_data["notify_data"]["body"]["services"][0]["event_time"]
            dt_object = datetime.strptime(event_time, "%Y-%m-%d %H:%M:%S")
            timestamp = int(dt_object.timestamp())

            properties = dict_data["notify_data"]["body"]["services"][0]["properties"]
            # print(event_time,timestamp,properties)

            device_id = dict_data["notify_data"]["header"]["device_id"]
            properties["device_id"] = device_id
            properties["event_time"] = timestamp
            properties["data_time"] = event_time
            # uuid 判断是否拥有
            c = mongdb_use()
            print("A,properties:", properties)
            c.insert_one_device(properties)

            # 调用数据库插入数据

        def thread_start():
            try:
                self.consumer_tag = self.channel.basic_consume(self.queue_name, on_message_callback=callback,
                                                               auto_ack=True,
                                                               consumer_tag=str(self.virtual_host + self.queue_name))

                while not self.stop_event.is_set():
                    if self.stop_event.is_set():
                        break
                    print("A, [*] Waiting for messages. To exit press CTRL+c")
                    print("consumer_tag:", self.virtual_host + self.queue_name)
                    self.channel.start_consuming()
            except pika.exceptions.ConnectionClosed:
                print("A,  Connection closed")

        if self.stop != "stop":
            self.receive_thread = threading.Thread(target=thread_start)
            self.receive_thread.start()

    def stop_accept(self):
        # def callback(ch, method, properties):
        #     self.channel.stop_consuming()
        self.stop = "stop"
        self.stop_event.set()  # 设置停止事件
        if self.channel.basic_consume:
            print("channel.basic_consume_cancel")
            self.channel.basic_cancel(consumer_tag=self.consumer_tag)
        if self.stop_event.is_set():
            print("channel")
        print("stop accept_join")
        self.receive_thread.join()  # 等待线程结束
        print("A, stop accepting connection")

# from pydantic import BaseSettings
#
#
# class Settings(BaseSettings):
#
#     mongo_uri: str = 'mongodb://root:123456@localhost:27017'
#
#
#
# settings = Settings()

class mongdb_use():
    def __init__(self, db_name="shebei", collection_name="all", clint_mongodb="localhost:27017"):
        self.client = pymongo.MongoClient("mongodb://" + clint_mongodb)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    def insert_one_device(self, dict_template, document=insert_template):
        # print("insert success", document)
        # print(document.format(**dict_template))
        # print(type(document.format(**dict_template)))
        list_docu = document.format(**dict_template).split(",")
        data_dict = {}
        print(list_docu)
        print(list_docu[0:-1])
        for pair in list_docu[0:-1]:
            # 使用冒号分割键和值
            key, value = pair.split(':')
            # 去除键和值两端的空格
            key = key.strip()
            value = value.strip()
            # 将键值对添加到字典中
            data_dict[key] = value
        print("1", data_dict)
        data_dict["data_time"] = list_docu[-1].replace("data_time:", "")
        print("data_dict:", data_dict)
        return self.collection.insert_one(data_dict)

    def insert_one_user(self, user_name, password, e_mail, name_number):
        return self.collection.insert_one(
            {"account": user_name, "passwd": generate_password_hash(password), 'e-mail': e_mail,
             "usrName": name_number, "date": int(time.time())})

    def insert_one_token(self, account, access_token, refresh_token):
        return self.collection.insert_one(
            {"account": account, "access_token": access_token, "refresh_token": refresh_token})

    def update_one_vcode(self, email, vcode, time_start, type):
        query = {"email": email, "type": type}
        new_values = {"$set": {"vcode": vcode, "time_start": time_start}}

        return self.collection.update_one(query, new_values)

    def insert_one_vcode(self, email, vcode, time_start, type):
        return self.collection.insert_one({"email": email, "vcode": vcode, "time_start": time_start, "type": type})

    def find_time(self, start_time, end_time):
        find_time_query = {
            "data_time": {
                '$gte': start_time,  # 大于等于开始时间
                '$lte': end_time  # 小于等于结束时间
            }
        }
        # 时间间隔最好在前端的缓存中去做
        print(find_time_query)
        return self.collection.find(query=find_time_query)

    def find_user_by_any(self, any, name="account"):
        find_user_query = {
            name: any
        }
        return dict(self.collection.find_one(find_user_query))

    def find_over_by(self, type, name, number):
        find_over_query1 = {
            name: {'$gt': number}
        }
        find_over_query2 = {
            name: {'$lt': number}
        }
        if type == "gt":
            return self.collection.find(find_over_query1)
        elif type == "lt":
            return self.collection.find(find_over_query2)

    def find_user_by_account(self, account_name):
        find_user_query = {
            "account": account_name
        }
        return self.collection.find_one(find_user_query)

    def find_user_by_id(self, id):
        find_user_query = {
            "_id": id
        }
        return self.collection.find_one(find_user_query)

    def find_vcode_email(self, email, type, vcode):
        find_vcode_email_query = {
            "email": email,
            "type": type,
            "vcode": vcode
        }
        return self.collection.find_one(find_vcode_email_query)

    def update_one_pass(self, query_find, query_var, update, update_var):
        query = {query_find: query_var}
        update_query = {"$set": {update: update_var}}
        return self.collection.update_one(query, update_query)

    def delete_one_vcode(self, query):
        de_vcode = {

        }
        return self.collection.delete_one(query)

    def close_connection(self):
        self.client.close()

    def connect_and_find(self):
        a = mongdb_use(db_name="user_test", collection_name="usr")


def check_value_over(name, type, number):
    a = mongdb_use(db_name="shebei", collection_name="all")
    documents = a.find_over_by(name=name, type=type, number=number)


class MyCustomException(Exception):
    def __init__(self, message):
        self.message = message
        super().__init__(self.message)


class Send_email():
    def __init__(self, smtp_server="smtp.qq.com", smtp_port=465, email_sender="",
                 email_password=""):  # 授权码
        # 设置SMTP服务器地址和端口
        self.smtp_server = smtp_server  # 你的SMTP服务器地址
        self.smtp_port = smtp_port  # SMTP服务器端口，通常为25, 465或587
        self.email_sender = email_sender
        self.email_password = email_password
        # 创建SMTP服务器的连接
        self.smtp = SMTP_SSL(self.smtp_server, self.smtp_port)

        # 启动TLS模式（如果服务器支持）
        # self.smtp.starttls()

    # def handle_insert(result, error=None):
    #     if error:
    #         print(f"Error inserting document: {error}")
    #     else:
    #         print(f"Document inserted with ID: {result.inserted_id}")
    def generate_verification_code(self, email, type, length=6, expiration_time=60):

        letters_and_digits = string.ascii_letters + string.digits
        code = ''.join((random.choice(letters_and_digits) for i in range(length)))
        time_now = int(time.time())
        c = mongdb_use(db_name="vcode", collection_name="user_vcode")
        c.insert_one_vcode(email=email, type=type, vcode=code, time_start=time_now)
        return code
        # try:
        #     c.insert_one_vcode(email=email,type=type,vcode=code,time_start=time_now)
        #     return code
        # except errors.DuplicateKeyError as e:
        #     self.handle_insert(None, str(e))

    def vcode_callback(self, vcode):
        return jsonify({"code": vcode})

    def find_vcode(self, email, type, vcode):
        c = mongdb_use(db_name="vcode", collection_name="user_vcode")
        ans = c.find_vcode_email(email=email, type=type, vcode=vcode)
        print("ans", ans)
        if not ans:
            print("vcode not right")
            raise MyCustomException("vcode not right")
            # return jsonify({'vcode': "not right"})
        else:
            time_now = int(time.time())
            if (time_now - ans["time_start"]) > 90:
                print(time_now - ans["time_start"])
                print("find_vcode timed out")
                # raise TimeoutError()
                raise MyCustomException("vcode_time_out")
                # return jsonify({'vcode':"time_out"})
            else:
                if vcode == ans["vcode"]:
                    print("vcode found")
                    return jsonify("send")

    def send(self, To, Subject, Body, From="yszbwqz@qq.com"):
        self.smtp.login(self.email_sender, self.email_password)
        self.From = From
        self.To = To
        self.Subject = Subject
        self.Body = Body
        # 登录到你的邮箱账户
        # email_sender = 'sender@example.com'  # 发件人邮箱地址
        # email_password = 'your_password'  # 发件人邮箱密码
        # smtp.login(email_sender, email_password)
        # with SMTP_SSL(self.smtp_server, self.smtp_port) as server:
        #     server.login(self.sender_email, self.password)
        #     server.sendmail(self.sender_email, receiver_email, msg.as_string())
        #     print("Email sent successfully through SSL!")
        # 创建邮件对象
        self.msg = MIMEMultipart()
        self.msg['From'] = self.From
        self.msg['To'] = self.To  # 收件人邮箱地址
        self.msg['Subject'] = self.Subject  # 邮件主题

        # 邮件正文
        # body = 'This is a test email sent from Python.'
        self.msg.attach(MIMEText(Body, 'plain'))

        # 发送邮件
        self.smtp.send_message(self.msg)
        self.smtp.quit()


class Flask_run():
    def __init__(self):
        self.app = Flask(__name__)
        self.app.config['JWT_SECRET_KEY'] = ''
        self.app.config['JWT_TOKEN_LOCATION'] = ['headers']  # 指定令牌可以从哪些位置获取
        self.app.config['JWT_REFRESH_TOKEN_ENABLED'] = True  # 启用刷新令牌
        self.app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(minutes=30)  # 设置访问令牌的过期时间
        self.app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)  # 设置刷新令牌的过期时间
        self.jwt = JWTManager(self.app)
        self.app.secret_key = ''
        self.signup()
        self.login()
        self.yanzheng()
        # self.logout()
        self.forget_pass()
        self.chongzhi_pass()
        self.task_results={}
        # self.chongzhi_pass()

    # def refesh(self):
    #     @self.app.route('/refresh', methods=['POST'])
    #     @jwt_required(refresh=True)
    #     def refresh():
    #         # 获取当前请求中的 JWT 标识（通常是用户 ID）
    #         identity = get_jwt_identity()
    #         # 创建新的访问令牌
    #         new_access_token = create_access_token(identity=identity)
    #         return jsonify(access_token=new_access_token)

    def register_routes(self):
        self.app.add_url_rule('/user/<username>', view_func=self.user_page, methods=['POST'])
        self.app.add_url_rule('/user_device', view_func=self.user_cl_device, methods=['POST'])
        self.app.add_url_rule('/user/time_mqtt', view_func=self.user_time_mqtt, methods=['POST'])
        self.app.add_url_rule('/user/ai/ai', view_func=self.user_ai, methods=['POST'])
        self.app.add_url_rule('/device_set_number', view_func=self.device_setnumber, methods=['POST'])
        # #通过flask启动websocket不是一个好选项

    # def red_all(self, url, code, da_ta):  #废弃的重定向 毕竟不是后端来做这个有问题
    #     # 设置重定向的目标路由
    #     redirect_url = url_for(url)
    #
    #     # 创建重定向响应
    #     response = make_response(redirect(redirect_url))
    #
    #     # 设置状态码
    #     response.status_code = 302
    #
    #     # 设置响应内容
    #     data = {
    #         'message': 'This is a JSON response'
    #     }
    #     response.data = jsonify(data).data
    #
    #     return response

    def get_user(self, user_name, name="account"):
        """根据用户名获得用户记录"""
        c = mongdb_use(db_name="user", collection_name="miss")
        user = c.find_user_by_any(name=name, any=user_name)
        print("getuser:{}-".format(name), user)
        return user
        # for user in USERS:
        #     if user.get("account") == user_name:
        #         print("get_user", user)
        #         return user
        # return None

    def signup(self):
        @self.app.route('/signup', methods=['POST'])
        def sign_view():
            print("flask signup")
            param = request.get_json()
            e_mail = param['email']
            name_number = param['usrName']  # 与密码下次一起post
            user_name = param['account']
            password = param['passwd']  # 等待下次post
            emsg = 'true'
            if user_name and password:
                user_info = self.get_user(user_name)  # 用用户名获取用户信息
                if user_info is None:
                    email_info = self.get_user(name="e-mail", user_name=e_mail)
                    # 邮件发送验证码
                    if email_info:
                        print("email_info:", email_info)
                        return jsonify({"error": "email already exists"})  # 邮箱已存在
                    else:
                        # 生成随机验证码

                        c = Send_email()
                        verification_code = c.generate_verification_code(email=e_mail, type="signup", )
                        body = f'Your verification code is: {verification_code}'
                        c.send(To=e_mail, Subject="验证邮箱", Body=body)
                else:
                    emsg = "usrExist"
                return emsg
        # return sign_view

    def yanzheng(self):
        @self.app.route('/signup/yanzheng', methods=['POST'])
        def yan_zheng_view():
            print("flask yanzhgeng")

            param = request.get_json()
            name_number = param['usrName']  # 与密码下次一起post
            user_name = param['account']
            password = param['passwd']  # 等待下次post

            e_mail = param['email']
            verification_code = param['verification_code']
            # print(verify_code_is_valid(e_mail, verification_code))

            # 检查邮件地址和验证码是否匹配（这里需要实现验证逻辑）

            try:
                b = Send_email()
                b.find_vcode(email=e_mail, type="signup", vcode=verification_code)
                # 验证码有效，继续注册流程或返回成功信息
                c = mongdb_use(db_name="user", collection_name="miss")
                c.insert_one_user(user_name, password, e_mail, name_number)
                return jsonify(message="验证码有效，注册成功！")
            except MyCustomException as e:
                return jsonify(f"捕获到自定义异常，错误信息是：{e.message}")

    def login(self):
        @self.app.route('/login', methods=['POST'])  # 登录
        def login_lo():
            print("flask login")
            param = request.get_json()
            user_name = param['account']
            password = param['passwd']

            emsg = {'msg': 'none'}
            if user_name and password:
                user_info = self.get_user(user_name=user_name)
                if user_info is None:
                    emsg = {'msg': 'usrNotExist'}
                else:
                    user = User(user_info)
                    if user.verify_password(password):
                        print("if", password)
                        c = mongdb_use(db_name="user", collection_name="miss")
                        msg = c.find_user_by_account(account_name=user_name)  # 返回字典
                        # usr_db = usr.find_one({"account": user_name}, {"_id": 1})
                        # usr_id = usr_db.get('_id')
                        emsg = {'msg': 'true'}

                        b = mongdb_use(db_name="token", collection_name="use")
                        print(user_name)
                        print(c.find_user_by_account(account_name=user_name))
                        # id = c.find_user_by_account(account_name=user_name)["_id"]
                        access_token = create_access_token(identity=str(user_name), fresh=True)
                        refresh_token = create_refresh_token(identity=str(user_name + password))
                        # self.login_successfully(user_id=msg["_id"])
                        # response=make_response(redirect(url_for('user_page',username=user_name,_external=True)))
                        # response.status_code=302
                        # response.data=jsonify(access_token=access_token, refresh_token=refresh_token)
                        jsonify_data = jsonify(
                            {"access_token": access_token, "refresh_token": refresh_token, "msg": "true"})
                        b.insert_one_token(account=user_name, access_token=access_token, refresh_token=refresh_token)
                        # print(jsonify_data)
                        return jsonify_data
                        # return jsonify(access_token=access_token, refresh_token=refresh_token)
                        # return emsg

                    else:
                        print("else", password)
                        emsg = {'msg': 'pwdErr'}
                result_json = json.dumps(emsg)
                return jsonify(result_json)
            else:
                print("flask without the passwod")

                result_json = json.dumps(emsg)
                return jsonify(result_json)
        # return login_lo

    @jwt_required(refresh=True)  # 绑定设备后进行记录
    def user_page(self, username):
        print("user_page:", username)
        param = request.get_json()
        status = param.get("status")
        print("Content-Type:", request.headers)
        # mqtt_run()

        # thread=threading.Thread(target=amqp_run)
        # thread.start()  #连接设备后调用amqp进行线程创建并一直运行
        if status:
            amqp_client = Amqp_accept(amqp_host='', amqp_port=5672, virtual_host="",
                                      amqp_username='', amqp_password='', amqp_queue_name="")
            amqp_client.stop = None
            amqp_client.connect()
            amqp_client.start_accept()
            if status == "stop":
                amqp_client.stop_accept()
        return jsonify("amqp_client start!"), 202

    # @jwt_required(refresh=True)
    def user_cl_device(self):
        param = request.get_json()
        device = param['device']
        device_status = param['device_status']
        client = Mqtt_client_test()
        client.connect()
        client.publish(topic=device, message=device_status)
        client.disconnect()
        return jsonify("device message send")
    @jwt_required(refresh=True)
    def device_setnumber(self):
        param = request.get_json()
        device = param['device']
        number= param['number']
        client = Mqtt_client_test()
        client.connect()
        client.publish(topic=device, message=number)
        client.disconnect()
        return jsonify("device message setnumber send")
    @jwt_required(refresh=True)
    def user_ai(self):
        print("Content-Type:", request.headers)
        param = request.get_json()
        ask = param['ask']
        return jsonify(get_baidu_json_make(get_baidu_response_json(ask)))

    @jwt_required(refresh=True)  # 定时和阈值控制或许可以一起
    def user_time_mqtt(self):
        to_ip_address =request.remote_addr
        param = request.get_json()
        print("user_time_mqtt",param)
        device = param['device']
        device_status_start = param['device_status_start']
        device_status_end = param['device_status_end']
        time_start = param['time_start']
        data_time_start = datetime.strptime(time_start, "%Y-%m-%d %H:%M:%S")
        data_time_start_tuple = (data_time_start.year, data_time_start.month, data_time_start.day
                                 , data_time_start.hour, data_time_start.minute, data_time_start.second)
        data_time_start_end = datetime(*data_time_start_tuple)
        time_end = param['time_end']
        data_time_end = datetime.strptime(time_end, "%Y-%m-%d %H:%M:%S")
        data_time_end_tuple = (data_time_end.year, data_time_end.month, data_time_end.day
                               , data_time_end.hour, data_time_end.minute, data_time_end.second)
        data_time_end_end = datetime(*data_time_end_tuple)

        def time_pulish(topic, message):
            client = Mqtt_client_test()
            client.connect()
            client.publish(topic=topic, message=message)
            client.disconnect()
            self.task_results[data_time_start_end+device] =data_time_start_end+device

        def time_start1(trigger_time, topic, message):
            scheduler = BackgroundScheduler()
            scheduler.add_job(time_pulish, trigger=DateTrigger(run_date=trigger_time),args=[ topic, message])
            scheduler.start()

        time_start1(trigger_time=data_time_start_end, topic=device, message=device_status_start)
        time_start1(trigger_time=data_time_end_end, topic=device, message=device_status_end)

        return jsonify({'message': '任务已添加到调度器'}), 202

    @jwt_required(refresh=True)
    def time_to_info(self):
        param = request.get_json()
        task_id=param['task_id']
        result = self.task_results.get(task_id)
        if result:
            del self.task_results[task_id]  # 删除已获取的结果，避免重复返回
            return jsonify({'status': 'completed', 'result': result})
        else:
            return jsonify({'status': 'pending'}), 200  # 任务还在等待或未开始
    def logout(self, request):
        @self.app.route('/logout')  # 登出
        @login_required
        def logout_lo():
            logout_user()
            return redirect(url_for('login'))

    def forget_pass(self):  # 需要增加邮箱重置的or方法
        @self.app.route('/forget', methods=['POST'])
        def forget_lo():
            print("flask forget password")  # 先接受账户 通过账户返回邮箱提升
            params = request.get_json()
            user_name = params['account']
            e_mail = params['email']  # 验证邮箱地址在进入重置密码
            c = mongdb_use(db_name="user", collection_name="miss")
            y = c.find_user_by_account(account_name=user_name)
            if not y:
                return json.dumps({'without the user_account'})
            else:
                email = y['email']
                if email != e_mail:
                    return jsonify('the e_mail is wrong')
                else:
                    c = Send_email()
                    verification_code = c.generate_verification_code(email=email, type="forget_pass")
                    body = f'Your verification code is: {verification_code}'
                    c.send(To=e_mail, Subject="重置密码", Body=body)
                    return jsonify({'send': True})

    def yanzheng_pass(self):
        @self.app.route('/forget/yanzheng', methods=['POST'])
        def yanzheng_pass_lo():
            params = request.get_json()
            verification_code = params['verification_code']
            c = Send_email()
            c.find_vcode(email=params['email'], type="forget_pass", vcode=verification_code)

    def chongzhi_pass(self):
        @self.app.route('/forget/chongzhi', methods=['POST'])
        def chongzhi_lo():
            print("flask chongzhi password")
            params = request.get_json()
            user_name = params['account']
            password = params['password']
            c = mongdb_use(db_name="user", collection_name="miss")
            c.update_one_pass(query_find="user_name", query_var=user_name, update="password", update_var=password)
            return json.dumps('成功修改')

    # def contaller_devices(self):   #废弃特性 有替代品了
    #     @self.app.route('/login/publish_message', methods=['POST'])
    #     def publish_message():
    #         print("flask publish_message")
    #         param = request.get_json()
    #         topic = param['topic']
    #         message = param['message']
    #         c = Mqtt_client_test()
    #         c.publish(topic, message)


#
# login_manager = LoginManager()  # 实例化登录管理对象
# login_manager.init_app(app)  # 初始化应用
# login_manager.session_protection = 'strong'
# login_manager.login_view = 'login'  # 设置用户登录视图函数 endpoint
#


#
# def get_user(user_name):
#     """根据用户名获得用户记录"""
#     for user in USERS:
#         if user.get("account") == user_name:
#             print("get_user", user)
#             return user
#     return None
#
#
class User(UserMixin):
    """用户类"""

    def __init__(self, user):
        self.username = user.get("account")
        self.password_hash = user.get("passwd")
        self.id = user.get("_id")

    def verify_password(self, password):
        """密码验证"""
        if self.password_hash is None:
            return False
        return check_password_hash(self.password_hash, password)

    def get_id(self):
        """获取用户ID"""
        return self.id

    @staticmethod
    def get(user_id):
        """根据用户ID获取用户实体，为 login_user 方法提供支持"""
        if not user_id:
            return None
        c = mongdb_use(db_name="user", collection_name="miss")
        msg = c.find_user_by_id(user_id)
        return User(msg)


# @login_manager.user_loader  # 定义获取登录用户的方法
# def load_user(user_id):
#     return User.get(user_id)


def mqtt_run():
    client = Mqtt_client_test()
    client.connect()
    # client.publish(topic="fengmingqi",message="ON")
    # client.publish(topic="shuibeng",message="ON")
    # client.publish(topic="fengshan", message="ON")
    print("1")
    # time.sleep(5)
    # print("4")
    # client.publish(topic="fengmingqi",message="OFF")
    # client.publish(topic="shuibeng", message="OFF")
    # client.publish(topic="fengshan", message="OFF")

    # 应该是做成单向在服务器上 只管publish


def amqp_run():
    amqp_client = Amqp_accept(amqp_host='117.78.16.25', amqp_port=5672, virtual_host="test35_C0rGb9aGXA",
                              amqp_username='apiuser', amqp_password='apipass', amqp_queue_name="test_hello")
    amqp_client.connect()
    amqp_client.start_accept()
    # 转发到库中在查找 应该一直开着 一个用户开一个线程也许？


class MyThread(threading.Thread):  # 未怎么使用的进程检查和纠错
    def __init__(self):
        super().__init__()
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def run(self):
        while not self._stop_event.is_set():
            print("Thread is running...")
            time.sleep(1)
        print("Thread is exiting...")


def check_and_restart_thread(thread):
    if not thread.is_alive():
        print("Thread is not running, restarting...")
        thread = MyThread()
        thread.start()
    else:
        print("Thread is running normally")


# 创建一个事件，用于指示服务器应该关闭
shutdown_event = asyncio.Event()


async def fetch_data_from_db(start_time, end_time, time_type):  # 异步查询时间内的数据
    # 连接MongoDB数据库
    client = AsyncIOMotorClient('mongodb://localhost:27017')
    db = client['shebei']
    collection = db['all']
    find_time_query = {
        "data_time": {
            '$gte': start_time,  # 大于等于开始时间
            '$lte': end_time  # 小于等于结束时间
        }
    }
    find_time_query1 = {
        "_id": {
            '$gte': start_time,  # 大于等于开始时间
            '$lte': end_time  # 小于等于结束时间
        }
    }
    # if time_type=="timestamp":
    #     cursor = collection.find(find_time_query1)
    # if time_type=="datatime":
    # # 查询数据库
    #     cursor = collection.find(find_time_query)
    cursor = collection.find(find_time_query1) if time_type == "timestamp" else collection.find(find_time_query)
    async for document in cursor:
        # print(document)
        # document 是一个字典，包含文档的字段名和字段值
        # 直接使用字典推导式来创建新的字典
        mapped_dict = {key: value for key, value in document.items()}
        data = mapped_dict['data_time']
        # class_type1 = int(mapped_dict[class_type])
        # yield mapped_dict
        # for key, value in mapped_dict:
        # yield data
        # yield class_type1
        # del mapped_dict[]
        yield mapped_dict
        # yield mapped_dict
    # data = await cursor.to_list(length=None) #废弃掉 列表不如生成式
    # return data


async def fetch_over_from_db(start_time, end_time, time_type, name, type, number):
    client = AsyncIOMotorClient('mongodb://localhost:27017')
    db = client['shebei']
    collection = db['all']

    # 连接MongoDB数据库

    # fetch_data_from_db(start_time, end_time, time_type)
    find_over_query1 = {
        name: {'$gt': number},
        "data_time": {
            '$gte': start_time,  # 大于等于开始时间
            '$lte': end_time  # 小于等于结束时间
        }
    }
    find_over_query2 = {
        name: {'$lt': number},
        "data_time": {
            '$gte': start_time,  # 大于等于开始时间
            '$lte': end_time  # 小于等于结束时间
        }
    }
    # if type == "gt":
    #     documents = collection.find(find_over_query1)
    # elif type == "lt":
    #     documents = collection.find(find_over_query2)
    documents = collection.find(find_over_query1) if type == "gt" else collection.find(find_over_query2)
    async for document in documents:
        # document 是一个字典，包含文档的字段名和字段值
        # 直接使用字典推导式来创建新的字典
        mapped_dict = {key: value for key, value in document.items()}
        count = int(len(mapped_dict))
        # over_dict={}
        # over_dict[name]=mapped_dict[name]
        # over_dict["time"]=mapped_dict["data_time"]
        # yield count
        yield mapped_dict


async def serverHands(websocket):
    while True:
        recv_text = await websocket.recv()
        print("recv_text=" + recv_text)
        if recv_text == "hello":
            print("connected success")

            await websocket.send("connected success")
            return True
        else:
            await websocket.send("connected fail")


# 接收从客户端发来的消息并处理，再返给客户端ok
async def serverRecv(websocket):
    while True:
        recv_text = await websocket.recv()
        print("re:", recv_text)
        recv_json = json.loads(recv_text)
        # mapped_dict_generator = fetch_data_from_db(start_time=recv_json["start_time"],end_time=recv_json["end_time"])  # 从MongoDB获取数据 #废弃掉 不能下文的生成式
        async for data in fetch_data_from_db(start_time=recv_json["start_time"],
                                             end_time=recv_json["end_time"],

                                             time_type=recv_json["time_type"]):  # 从MongoDB获取数据
            print(data)
            # for key, value in data:
            #     print(key, value)
            #     await websocket.send(key, value)
            # time.sleep(0.2)
            await websocket.send(json.dumps(data))
        # print(mapped_dict_generator)
        # print(type(mapped_dict_generator))
        # mapped_dict_list = [item async for item in mapped_dict_generator]
        # print("data:",mapped_dict_list)
        # print(type(mapped_dict_list))
        # json_str = json.dumps(mapped_dict_list)
        # await websocket.send(json_str)  # 将数据通过WebSocket发送
        # websocket.close()
        # await websocket_handler(start_time=recv_json["start_time"],end_time=recv_json["end_time"])
        # print("recv:", recv_text)
        # print(message_time)
        # await websocket.send(message_time)
        print("end")


async def server_over(websocket):
    while True:
        recv_text = await websocket.recv()
        print("re:", recv_text)
        recv_json = json.loads(recv_text)
        async for data in fetch_over_from_db(start_time=recv_json["start_time"],
                                             end_time=recv_json["end_time"],

                                             time_type=recv_json["time_type"],
                                             name=recv_json["name"], type=recv_json["type"],
                                             number=recv_json["number"]):  # 从MongoDB获取数据
            print(data)
            time.sleep(0.2)
            # data_over= "超出阈值"+str(data)
            await websocket.send(json.dumps(data))


# 握手并且接收数据
async def serverRun(websocket, path):
    print(path)
    await serverHands(websocket)

    await serverRecv(websocket)


async def serverRun1(websocket, path):
    print(path)
    await serverHands(websocket)

    await server_over(websocket)


async def run_asynico_loop():
    # loop = asyncio.new_event_loop()
    # asyncio.set_event_loop(loop)

    try:
        server = websockets.serve(serverRun, "", 8765)
        await  asyncio.get_event_loop().run_until_complete(server)  # 运行服务器等待准备
        # 运行事件循环直到被关闭事件中断
        # asyncio.get_event_loop().run_until_complete(shutdown_event.wait())
        await  asyncio.get_event_loop().run_until_complete(shutdown_event.wait())
    except KeyboardInterrupt:
        server.send("closed by hand in the device")
        print("interrupt")
    finally:
        server.close()
        asyncio.get_event_loop().run_until_complete(server.wait_closed())
        asyncio.get_event_loop().stop()


# Dengdai

# server = websockets.serve(serverRun, "192.168.31.111", 8765)
# asyncio.get_event_loop().run_until_complete(server)  # 运行服务器等待准备
# try:
#     # 运行事件循环直到被关闭事件中断
#     asyncio.get_event_loop().run_until_complete(shutdown_event.wait())
#     # asyncio.get_event_loop().run_forever()
# except KeyboardInterrupt:
#     server.send("closed by hand in the device")
#     print("interrupt")
# finally:
#     server.close()
#     asyncio.get_event_loop().run_until_complete(server.wait_closed())
#     asyncio.get_event_loop().stop()


def flask_run():
    print("Starting flask server...")
    c = Flask_run()
    c.register_routes()
    c.app.run(host='0.0.0.0', port=5000)


if __name__ == "__main__":
    # threading1 = threading.Thread(target=mqtt_run)
    # threading2 = threading.Thread(target=amqp_run)
    # threading3 = threading.Thread(target=run_asynico_loop)
    # threading3.start()
    threading4 = threading.Thread(target=flask_run)
    # threading1.start()
    # threading2.start()

    threading4.start()
    try:
        server = websockets.serve(serverRun, "0.0.0.0", 8765)
        # server = websockets.serve(serverRun, "192.168.43.150", 8765)
        asyncio.get_event_loop().run_until_complete(server)  # 运行服务器等待准备
        # 运行事件循环直到被关闭事件中断
        server1 = websockets.serve(serverRun1, "0.0.0.0", 8766)
        asyncio.get_event_loop().run_until_complete(server1)
        # asyncio.get_event_loop().run_until_complete(shutdown_event.wait())
        asyncio.get_event_loop().run_until_complete(shutdown_event.wait())
        asyncio.get_event_loop().run_until_complete(shutdown_event.wait())
    except KeyboardInterrupt:
        # server.send("closed by hand in the device")
        print("interrupt")
    # finally:
    #     server.close()
    #     asyncio.get_event_loop().run_until_complete(server.wait_closed())
    #     asyncio.get_event_loop().stop()
    # asyncio.run(run_asynico_loop())

    # threading1.join()
    # threading2.join()

    threading4.join()

    # loop = asyncio.new_event_loop()

    # threading3.join()
    # 创建并启动线程
    # thread = MyThread()
    # thread.start()
    #
    # # 定期检查线程状态并重启
    # while True:
    #     check_and_restart_thread(thread)
    #     time.sleep(5)  # 每5秒检查一次
print("Finished")
